/***************************************************************************************
  *Copyright(C),2010-2016,Sumscope
  *FileName   : ThreadSorter.cpp
  *Author     : scofined.qi
  *Version    : 1.0
  *Date       : 2016/06/04
  *Desc       : //用于说明此程序文件完成的主要功能
  *Relation   : 
  *Other      : //其他内容说明
  *Function   : //主要函数列表，每条记录应包含函数名及功能简要说明
  *History    : //修改历史纪录列表，每条修改记录应包含修改日期，修改者以及修改内容简介
***************************************************************************************/

#include "ThreadSorter.h"
#include "TableUser.h"
#include "Table.h"
#include "ThreadPool.h"
#include "SortBuffer.h"
#include "SortTask.h"
#include <core/time/time.h>
#include <QThread>

#ifdef WIN32
#define BASSERT assert
#else
#define BASSERT //
#endif

using namespace qb::base;
namespace qb{
	namespace ht{
bool ThreadSorter::SortRangeStack::Pop(SORTPOS& q)
{
	if (m_count > 0)
	{
		q = m_ranges.at(m_count - 1);
		m_count--;
		return true;
	}
	return false;
}

void ThreadSorter::SortRangeStack::Add(int begin, int end)
{
	while (m_count >= (int)m_ranges.size()){ m_ranges.push_back(SORTPOS()); }
	SORTPOS& st = m_ranges.at(m_count); m_count++;
	st.left = begin;
	st.right = end;
}
//////////////////////////////////////////////////////////////////////////
ThreadSorter::ThreadSorter() :
sort_(0),
m_temp_size(0),
m_comp_temp(0),
granularity_(1024),
first_granul_(1024),
range_(0),
comp_count_(0),
swap_count_(0),
m_task_cache(0),
m_thread_pool(0)
{
	m_task_cache = new TaskCache();
}

ThreadSorter::~ThreadSorter()
{
	if (m_comp_temp){ delete[] m_comp_temp; m_task_cache = 0; }
	if (m_task_cache){ delete m_task_cache; m_task_cache = 0; }
	if (m_thread_pool){ delete m_thread_pool; m_thread_pool = 0; }
}

//////////////////////////////////////////////////////////////////////////

bool ThreadSorter::PrepareTempBuffer(int nsize)
{
    BASSERT(nsize > 4);
	if (m_temp_size < nsize || m_comp_temp == 0)
	{
		if (m_comp_temp)	delete[] m_comp_temp;
		m_comp_temp = new char[nsize + 1];
		memset(m_comp_temp, 0, nsize + 1);
	}
	m_record_size = nsize;
	m_temp_size = m_temp_size<nsize ? nsize : m_temp_size;
    BASSERT(m_temp_size >= nsize);
    BASSERT(m_comp_temp);
	return true;
}


void ThreadSorter::ThreadSort(void* base, unsigned ncount, unsigned w, FuncCompColumn comp_func_)
//void ThreadSorter::ThreadSort(char* a, int ncount,int w, FuncCompColumn comp_func_)
{
	//__try{
		char* a = (char*)base;
		//PrepareTempBuffer(w);

		// 排序复用，排序缓存
		// 当两次排序的sortbuffer相同的时候，将采用相同的结果
		// 首先得 判断相同的场景是否普遍
		sort_ = (unsigned char*)a;

		int p = 0, r = ncount - 1;
		range_ = (r + 1 - p);
		granularity_ = first_granul_ = 1024;
		m_task_cache->Reset();
        char* temp = (char*)alloca(m_record_size + 1);
		memset(temp, 0, m_record_size + 1);
		// 此段代码放置倒叙排列,快排对于倒叙排列的数据,速度相当慢
		// 如果一个数列倒序排列,那么使用通常的快排,将是非常缓慢
		// 此段代码首先检测倒序程度,超过70%的倒序,就点到数列;
		
		int direction = 0;
		if (1)
		{
			for (int i = p + 1; i <= r; i++){
				int cmpret = comp_func_(this, &a[i], &a[(i - 1)]);
				direction += cmpret > 0 ? 1 : (cmpret < 0 ? -1 : 0);
			}
			if (direction < (int)(range_*(-0.2))){
				for (unsigned int i = p*w, j = r*w; i < j; i += w, j -= w)
					SwapMemory(&a[i], &a[j], w);
			}
		}

		{
			GetThreadPool()->ChangeMode(kThreadModeFast);
			SplitTask(a, w, p, r, comp_func_);
			SortTask* first_task = m_task_cache->GetFirstTask();
			//if (first_task)
			//{
			//	CMETRICS("ThreadSortFunc(first_task)");
			//	ThreadSortFunc(first_task);
			//}
				
		}


		// 等待后台所有线程结束完毕
		{
			m_task_cache->WaitAllFinish();
		}
		GetThreadPool()->ChangeMode(kThreadModeSlow);
		//CheckSortResult();
	//}
	//__except (1)
	//{
	//	DBGVIEW("ThreadSort has exception");
	//}

}
ThreadPool* ThreadSorter::GetThreadPool()
{
	if (m_thread_pool == 0)
		m_thread_pool = new ThreadPool();
	return m_thread_pool;
}
void ThreadSorter::AddTask(char* a, int w, unsigned int p, unsigned int r, FuncCompColumn comp)
{
	ThreadPool* pool = GetThreadPool();
	SortTask* task = m_task_cache->GetTask(this, a, w, p, r, comp);

	// 任务分配策略,通过调整granularity_来控制任务
	// 自动根据排序规模约定每个排序任务的最少排序记录个数，任务粒度，不少于first_granul_
	if (m_task_cache->GetWaitTaskCount() <= 0)
	{
		//granularity_ = range_ / pool->GetThreadCount();
		//granularity_ = granularity_ < first_granul_ ? first_granul_ : granularity_;
	}

	// 最开始的一段交给主线程来处理
	//if (p != 0)
	pool->PostTask(ThreadSortFunc, task);

	// 任务统计,稍后主线程来检查完成时刻
	m_task_cache->AddWaitTask(task, p == 0);
}

void ThreadSorter::SplitTask(char* a, int w, unsigned int p, unsigned int r, FuncCompColumn comp_func_)
{
	// 主线程负责将单条记录大小为w的数组a的范围[p,r]划分为三部分
	// 其中[p,q.left-1]是小的
	// [q.right+1,r]是大的
	// [q.left,q.right]是相等的，不需要排序
	if (1)
	{
		m_stack.Clear();
		m_stack.Add(p, r);
		SORTPOS range, q;
		static double part = 0, total = 0;
		static int count = 0;

		granularity_ = (r + 1 - p) / (2*GetThreadPool()->GetThreadCount());
		HighResTime sumtimer;
		for (;;)
		{
			if (!m_stack.Pop(range))
				break;
			if (range.left >= range.right)
				continue;
			HighResTime timer;
			q = Partition(a, w, range.left, range.right, comp_func_);
					part += timer.elapsed();
			if (q.left - range.left <= granularity_)
				AddTask(a, w, range.left, q.left - 1, comp_func_);
			else
				m_stack.Add(range.left, q.left - 1);

			if (range.right - q.right <= granularity_)
				AddTask(a, w, q.right + 1, range.right, comp_func_);
			else
				m_stack.Add(q.right + 1, range.right);
		}

		//count++;
		//total += sumtimer.elapsed().m_time;
		//static DWORD last = GetTickCount();
		//if (GetTickCount() - last >= 1000)
		//{
		//	last = GetTickCount();
		//	DebugView(0, 0, 0, "[ThreadSorter::SplitTask]:Partition::%.6f  Total:%.6f percent:%.2f  each:%.6f",
		//		part, total, part * 100 / total,part/count);
		//}


	}
	else
	{
		if (p >= r)
			return;
		SORTPOS q = Partition(a, w, p, r, comp_func_);

		if (q.left - (int)p <= granularity_)
			AddTask(a, w, p, q.left - 1, comp_func_);
		else
			SplitTask(a, w, p, q.left - 1, comp_func_);

		if ((int)r - q.right <= granularity_)
			AddTask(a, w, q.right + 1, r, comp_func_);
		else
			SplitTask(a, w, q.right + 1, r, comp_func_);
	}


}
SORTPOS ThreadSorter::Partition(char* arr, int width, int left, int right, FuncCompColumn comp_func_)
{
	// 此函数能够对付数据量重复较多的数列;
	int i, j;
	int pleft, pright;
	int tag;
	SORTPOS p;

	i = left - 1, j = right;
	pleft = left - 1, pright = right;
	int mid = left + (right + 1 - left) / (2);

			//int a, b, c;
	char* mi = &arr[left], *mx = &arr[mid], *md = &arr[right];
	// 使得arr[right]成为中值
	if (comp_func_(this, &arr[left], &arr[mid]) > 0)
	{
		mi = &arr[mid], mx = &arr[left];
	}
	if (comp_func_(this, mi, &arr[right]) > 0)md = mi;
	else if (comp_func_(this, &arr[right], mx) > 0)md = mx;
	else	md = &arr[right];

	if (md != &arr[right])SwapMemory(md, &arr[right], width);
	//Exchange(&arr[mid], &arr[right]);
    char* base = /*m_comp_temp; */(char*)alloca(width + 1); base[width] = 0;
	memcpy(base, &arr[right], width);
	//@_1   
	for (;;)
	{
		// 比i小的都肯定比base小
		while (i < j && comp_func_(this, &arr[(++i)], base) < 0);

		// 比j大的都肯定比base大
		while (i < j && comp_func_(this, &arr[(--j)], base) > 0) if (j <= left) break;

		if (j == i) break;

		// i已经不比base小,j已经不比base大了,所以需要交换
		SwapMemory(&arr[j], &arr[i], width); // @_2  

		// 将与base相等的依次拷贝到数列左侧
		if (comp_func_(this, &arr[i], base) == 0)
		{
			pleft += 1;
			SwapMemory(&arr[i], &arr[pleft], width);
		}

		// 将与base相等的依次拷贝到数列的右侧
		if (comp_func_(this, &arr[j], base) == 0)
		{
			pright -= 1;
			SwapMemory(&arr[j], &arr[pright], width);//@_3  
		}
	}//for  i==j

	// 经过上述处理,数列情况如下:
	//arr[left,pleft) 都与base相等	arr[pleft,i) 都比base小
	//arr(pright,right] 都与base相等	arr(j,pright] 都比base大

	int pos = i;
	SwapMemory(&arr[right], &arr[i], width); //@_4  

	j = i - 1; i = i + 1;

	// 处理重复数据的问题
	// 将与base相等的转移到数组中间
	for (tag = left; tag <= pleft; tag++, j--) SwapMemory(&arr[tag], &arr[j], width);

	// 将与base相等的原先放在数列右端的移到中间
	for (tag = right - 1; tag >= pright; tag--, i++) SwapMemory(&arr[tag], &arr[i], width);//@_5  

	// j 是划分好的左侧数据的右端,i是划分好的右侧数据的左端,[j+1,i-1] 都是与base相等的重复数列
	// j>=left-1,i<=right+1
	p.left = j + 1, p.right = i - 1;
	return p;
}

// 比较好的交换内存的宏，来自redis
/*
* Qsort routine from Bentley & McIlroy's "Engineering a Sort Function".
*/
#define swapcode(TYPE, parmi, parmj, n) { 		\
	size_t i = (n) / sizeof (TYPE); 		\
	TYPE *pi = (TYPE *)(void *)(parmi); 		\
	TYPE *pj = (TYPE *)(void *)(parmj); 		\
	do { 						\
		TYPE	t = *pi;			\
		*pi++ = *pj;				\
		*pj++ = t;				\
	        } while (--i > 0);				\
}

#define SWAPINIT(a, es) swaptype = ((char *)a - (char *)0) % sizeof(long) || \
	es % sizeof(long) ? 2 : es == sizeof(long)? 0 : 1;

// 要求 w%4==0
#define SWAPUINT(a,b,w)	{							\
	unsigned int c = 0;								\
	unsigned int* src = (unsigned int*)a;			\
	unsigned int* dst = (unsigned int*)b;			\
	unsigned int* end = src + (width >> 2);			\
	while (src < end)								\
	{												\
		c = *src; *src = *dst; *dst = c;			\
		++src, ++dst;;								\
	}}

// 要求 w%8==0
#define SWAPUINT64(a,b,w)	{						\
    uint64_t c = 0;							\
    uint64_t* src = (uint64_t*)a;	\
    uint64_t* dst = (uint64_t*)b;	\
    uint64_t* end = src + (width >> 3);		\
	while (src < end)								\
	{												\
		c = *src; *src = *dst; *dst = c;			\
		++src, ++dst;;								\
	}}


#define SWAPMEMCPY(a,b,w,t)	{						\
	memcpy(t, a, w);								\
	memcpy(a, b, w);								\
	memcpy(b, t, w);								\
	}
	

inline void ThreadSorter::SwapMemory(char *a, char *b, unsigned int width){
	// a,b为两条记录的起始地址
	// width为一条记录的内存大小
	// temp为其提供交换的临时内存

	// sort_ 为数组的起始内存地址
	// width 为数组的每一个数据项的所占据的内存字节个数,只可能是1,2,4,8
	// width 只能是 0,1,2,3,
	// memcpy 已经非常快,经测试几乎不可优化
	//SWAPMEMCPY(a, b, width, temp);
	SWAPUINT64(a, b, width);
	//SWAPUINT(a, b, width);


	//SASSERT(width % 8 == 0);
	//unsigned __int64 c = 0;
	//unsigned __int64* src = (unsigned __int64*)a;
	//unsigned __int64* dst = (unsigned __int64*)b;
	//unsigned __int64* end = src + (width >> 3);
	//while (src < end)
	//{
	//	c = *src; *src = *dst; *dst = c;
	//	++src, ++dst;;
	//}


	// 假设width是4的倍数
	//SASSERT(width % 4 == 0);
	//unsigned int c = 0;
	//unsigned int* src = (unsigned int*)a;
	//unsigned int* dst = (unsigned int*)b;
	//unsigned int* end = src + (width >> 2);
	//while (src < end)
	//{
	//	c = *src; *src = *dst; *dst = c;
	//	++src, ++dst;;
	//}


	//int len = width >> 2;
	//for (int i = 0; i < len; ++i, a += 4, b += 4)
	//{
	//	c = *((unsigned int*)a);;
	//	*((unsigned int*)a) = *((unsigned int*)b);
	//	*((unsigned int*)b) = c;
	//}

	// 测试表明自行写的比memcpy慢一半速度
	//char c;
	//for (int i = 0; i < width; i++)
	//{
	//	c = *a; *a = *b; *b = c;
	//}
//#ifdef _DEBUG
	//swap_count_++;
//#endif
}
void ThreadSorter::ShortSort(char *lo, char *hi, unsigned width,
	FuncCompColumn comp){
	char *p, *max;
	while (hi > lo) {
		/* A[i] <= A[j] for i <= j, j > hi */
		max = lo;
		for (p = lo + width; p <= hi; p += width) {
			/* A[i] <= A[max] for lo <= i < p */

			if (comp(this, p, max) > 0) {
				max = p;
			}
			/* A[i] <= A[max] for lo <= i <= p */
		}

		/* A[i] <= A[max] for lo <= i <= hi */

		SwapMemory(max, hi, width);

		/* A[i] <= A[hi] for i <= hi, so A[i] <= A[j] for i <= j, j >= hi */

		hi -= width;

		/* A[i] <= A[j] for i <= j, j > hi, loop top condition established */
	}
}
#define CUTOFF 8            /* testing shows that this is good value */
#define STKSIZ (8*sizeof(void*) - 2)

// 功能:对所有记录的字段进行排序
// 线程:GUI线程
// base		是所有记录的相同字段组成的数组,数据项包括char,short,int,long,float,double,char* 等
// num		所有记录的个数,个数可能达100w数量级
// width	该数组的数据项的宽度,不会超过8

void ThreadSorter::QuickSort(
	void *base,
	unsigned num,
	unsigned width,
	FuncCompColumn comp)
{
	// 每个子线程对分配到的一个区间进行排序
	// 此部分代码是原始的qsort代码

	char *lo, *hi;              /* ends of sub-array currently sorting */
	char *mid;                  /* points to middle of subarray */
	char *loguy, *higuy;        /* traveling pointers for partition step */
	size_t size;                /* size of the sub-array */
	char *lostk[STKSIZ], *histk[STKSIZ];
	int stkptr;                 /* stack for saving sub-array to be processed */
	/* validation section */
	if (base == NULL || num <2)
		return;

	if (width <= 0)
		return;

	if (num < 2)
		return;                 /* nothing to do */

	stkptr = 0;                 /* initialize stack */

	lo = (char *)base;
	hi = (char *)base + width * (num - 1);        /* initialize limits */

	/* this entry point is for pseudo-recursion calling: setting
	lo and hi and jumping to here is like recursion, but stkptr is
	preserved, locals aren't, so we preserve stuff on the stack */
recurse:

	size = (hi - lo) / width + 1;        /* number of el's to sort */

	/* below a certain size, it is faster to use a O(n^2) sorting method */
	if (size <= CUTOFF) {
		ShortSort(lo, hi, width, comp);
	}
	else {
		/* First we pick a partitioning element.  The efficiency of the
		algorithm demands that we find one that is approximately the median
		of the values, but also that we select one fast.  We choose the
		median of the first, middle, and last elements, to avoid bad
		performance in the face of already sorted data, or data that is made
		up of multiple sorted runs appended together.  Testing shows that a
		median-of-three algorithm provides better performance than simply
		picking the middle element for the latter case. */

		mid = lo + (size / 2) * width;      /* find middle element */

		/* Sort the first, middle, last elements into order */
		if (comp(this, lo, mid) > 0) {
			SwapMemory(lo, mid, width);
		}
		if (comp(this, lo, hi) > 0) {
			SwapMemory(lo, hi, width);
		}
		if (comp(this, mid, hi) > 0) {
			SwapMemory(mid, hi, width);
		}

		/* We now wish to partition the array into three pieces, one consisting
		of elements <= partition element, one of elements equal to the
		partition element, and one of elements > than it.  This is done
		below; comments indicate conditions established at every step. */

		loguy = lo;
		higuy = hi;

		/* Note that higuy decreases and loguy increases on every iteration,
		so loop must terminate. */
		for (;;) {
			/* lo <= loguy < hi, lo < higuy <= hi,
			A[i] <= A[mid] for lo <= i <= loguy,
			A[i] > A[mid] for higuy <= i < hi,
			A[hi] >= A[mid] */

			/* The doubled loop is to avoid calling comp(mid,mid), since some
			existing comparison funcs don't work when passed the same
			value for both pointers. */

			if (mid > loguy) {
				do  {
					loguy += width;
				} while (loguy < mid && comp(this, loguy, mid) <= 0);
			}
			if (mid <= loguy) {
				do  {
					loguy += width;
				} while (loguy <= hi && comp(this, loguy, mid) <= 0);
			}

			/* lo < loguy <= hi+1, A[i] <= A[mid] for lo <= i < loguy,
			either loguy > hi or A[loguy] > A[mid] */

			do  {
				higuy -= width;
			} while (higuy > mid && comp(this, higuy, mid) > 0);

			/* lo <= higuy < hi, A[i] > A[mid] for higuy < i < hi,
			either higuy == lo or A[higuy] <= A[mid] */

			if (higuy < loguy)
				break;

			/* if loguy > hi or higuy == lo, then we would have exited, so
			A[loguy] > A[mid], A[higuy] <= A[mid],
			loguy <= hi, higuy > lo */

			SwapMemory(loguy, higuy, width);

			/* If the partition element was moved, follow it.  Only need
			to check for mid == higuy, since before the swap,
			A[loguy] > A[mid] implies loguy != mid. */

			if (mid == higuy)
				mid = loguy;

			/* A[loguy] <= A[mid], A[higuy] > A[mid]; so condition at top
			of loop is re-established */
		}

		/*     A[i] <= A[mid] for lo <= i < loguy,
		A[i] > A[mid] for higuy < i < hi,
		A[hi] >= A[mid]
		higuy < loguy
		implying:
		higuy == loguy-1
		or higuy == hi - 1, loguy == hi + 1, A[hi] == A[mid] */

		/* Find adjacent elements equal to the partition element.  The
		doubled loop is to avoid calling comp(mid,mid), since some
		existing comparison funcs don't work when passed the same value
		for both pointers. */

		higuy += width;
		if (mid < higuy) {
			do  {
				higuy -= width;
			} while (higuy > mid && comp(this, higuy, mid) == 0);
		}
		if (mid >= higuy) {
			do  {
				higuy -= width;
			} while (higuy > lo && comp(this, higuy, mid) == 0);
		}

		/* OK, now we have the following:
		higuy < loguy
		lo <= higuy <= hi
		A[i]  <= A[mid] for lo <= i <= higuy
		A[i]  == A[mid] for higuy < i < loguy
		A[i]  >  A[mid] for loguy <= i < hi
		A[hi] >= A[mid] */

		/* We've finished the partition, now we want to sort the subarrays
		[lo, higuy] and [loguy, hi].
		We do the smaller one first to minimize stack usage.
		We only sort arrays of length 2 or more.*/

		if (higuy - lo >= hi - loguy) {
			if (lo < higuy) {
				lostk[stkptr] = lo;
				histk[stkptr] = higuy;
				++stkptr;
			}                           /* save big recursion for later */

			if (loguy < hi) {
				lo = loguy;
				goto recurse;           /* do small recursion */
			}
		}
		else {
			if (loguy < hi) {
				lostk[stkptr] = loguy;
				histk[stkptr] = hi;
				++stkptr;               /* save big recursion for later */
			}

			if (lo < higuy) {
				hi = higuy;
				goto recurse;           /* do small recursion */
			}
		}
	}

	/* We have sorted the array, except for any pending sorts on the stack.
	Check if there are any, and do them. */

	--stkptr;
	if (stkptr >= 0) {
		lo = lostk[stkptr];
		hi = histk[stkptr];
		goto recurse;           /* pop subarray from stack */
	}
	else
		return;                 /* all subarrays done */
}
void ThreadSorter::ThreadSortFunc(void* p)
{
	// 排序子线程的任务入口
	SortTask* task = (SortTask*)p;
    task->threadid = (unsigned long)QThread::currentThreadId;
    task->timepoints[kTaskTimeBegin] = HighResTime();
	char* base = task->base + (task->left*task->width);
	int num = task->right + 1 - task->left;
	task->sorter->QuickSort(base, num, task->width, task->comp_func);
			task->timepoints[kTaskTimeFinish] = HighResTime();
	task->finished = true;
}
int ThreadSorter::StaticCompareColumns(ThreadSorter* ts, const void* pSortData1, const void* pSortData2)
{
	return ts->CompareColumns(pSortData1, pSortData2);
}

int ThreadSorter::StaticCompareColumns2(void* context, const void* pSortData1, const void* pSortData2)
{
	ThreadSorter* ts = (ThreadSorter*)context;
	return ts->CompareColumns(pSortData1, pSortData2);
}

void ThreadSorter::CheckSortResult()
{
	for (int i = 1; i < range_; i++)
	{
		unsigned char* cur = sort_ + m_record_size*i;
		int cmp = CompareColumns(cur - m_record_size, cur);
	}
}
	}//namespace ht
}//namespace ss
